//
// Created by HhL on 2020/12/24.
//

#include "thread_task_pool.h"
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <assert.h>
#include "common.h"

jint thread_pool_init(JNIEnv *jEnv, jint minSize, jint maxSize) {
    /* 为线程池申请内存 */
    g_pthread_pool = calloc(1, sizeof(thread_pool));
    if (NULL == g_pthread_pool) {
        throwRuntimeException(jEnv, "为线程池分配内存失败");

        goto return1;
    }
    /* 为工作线程集合分配内存 */
    g_pthread_pool->worker_tids = calloc(maxSize, sizeof(pthread_t));
    if (NULL == g_pthread_pool->worker_tids) {
        throwRuntimeException(jEnv, "为工作线程集合分配内存失败");

        goto return2;
    }

    /* 初始化互斥锁 */
    if (0 != pthread_mutex_init(&g_pthread_pool->lock, NULL)) {
        throwRuntimeException(jEnv, "初始化互斥锁失败");

        goto return3;
    }

    g_pthread_pool->thr_min_size = minSize;
    g_pthread_pool->thr_max_size = maxSize;
    g_pthread_pool->thr_live_size = minSize;
    g_pthread_pool->thr_busy_size = 0;

    return 0;
    return3:
    free(g_pthread_pool->worker_tids);
    g_pthread_pool->worker_tids = NULL;
    return2:
    free(g_pthread_pool);
    g_pthread_pool = NULL;
    return1:
    return -1;
}

void thread_pool_destroy() {
    assert(g_pthread_pool != NULL);
    for (int i = 0; i < g_pthread_pool->thr_live_size; ++i) {
        pthread_join(g_pthread_pool->worker_tids[i], NULL);
    }

    thread_pool_free();
}

void thread_pool_free() {
    print_nobuffer("【线程池销毁程序】任务执行结束，开始释放资源\n");

    /* 先释放任务池资源 */
    task_pool_free();
    print_nobuffer("【销毁线程池】开始释放线程池资源\n");
    pthread_mutex_destroy(&g_pthread_pool->lock);
    free(g_pthread_pool->worker_tids);
    free(g_pthread_pool);

    g_pthread_pool = NULL;
}

void thread_pool_print(const char *msg) {
    assert(msg != NULL);

    char str[BUFSIZ] = {0};

    sprintf(str, "%s: min_size=%d, max_size=%d, live_size=%d, busy_size=%d\n",
            msg, g_pthread_pool->thr_min_size, g_pthread_pool->thr_max_size, g_pthread_pool->thr_live_size,
            g_pthread_pool->thr_busy_size);

    write(STDOUT_FILENO, str, strlen(str));
}

/**=====================================
 * 任务池相关操作函数
 ======================================*/
void *task_run(JNIEnv *jEnv, jclass taskClass) {
    assert(taskClass != NULL);

    JNIEnv env = *jEnv;

    /* 获取任务的run方法 */
    jmethodID runMethod = (*jEnv)->GetMethodID(jEnv, taskClass, "run", "()V");
    if (NULL == runMethod) {
        throwRuntimeException(jEnv, "任务类缺少run方法");

        goto return1;
    }
    jobject taskObj = env->AllocObject(jEnv, taskClass);
    if (NULL == taskObj) {
        throwRuntimeException(jEnv, "实例化任务对象失败");

        goto return1;
    }

    /* 执行运行函数 */
    env->CallVoidMethod(jEnv, taskObj, runMethod);

    /* 释放任务对象资源 */
    env->DeleteLocalRef(jEnv, taskObj);
    env->DeleteGlobalRef(jEnv, taskClass);

    return1:
    return NULL;
}

jint task_pool_init(JNIEnv *jEnv, jint maxSize) {
    g_ptask_pool = calloc(1, sizeof(task_pool));
    if (NULL == g_ptask_pool) {
        throwRuntimeException(jEnv, "为任务池分配内存失败");

        goto return1;
    }
    g_ptask_pool->task_cqueue = calloc(maxSize, sizeof(task_pool));
    if (NULL == g_ptask_pool->task_cqueue) {
        throwRuntimeException(jEnv, "为任务队列分配内存失败");

        goto return2;
    }
    if (0 != pthread_mutex_init(&g_ptask_pool->lock, NULL)) {
        throwRuntimeException(jEnv, "初始化任务池的锁失败");

        goto return3;
    }
    if (0 != pthread_cond_init(&g_ptask_pool->cond_empty, NULL)) {
        throwRuntimeException(jEnv, "初始化任务池的条件变量失败");

        goto return4;
    }

    g_ptask_pool->queue_max_size = maxSize;
    g_ptask_pool->queue_size = 0;
    g_ptask_pool->queue_current = 0;

    return 0;

    return4:
    pthread_mutex_destroy(&g_ptask_pool->lock);
    return3:
    free(g_ptask_pool->task_cqueue);
    g_ptask_pool->task_cqueue = NULL;
    return2:
    free(g_ptask_pool);
    g_ptask_pool = NULL;
    return1:
    return -1;
}

jint task_pool_add(JNIEnv *jEnv, jstring taskClassName) {
    assert(g_ptask_pool != NULL);
    assert(taskClassName != NULL);
    JNIEnv env = *jEnv;

    jclass taskClass = env->FindClass(jEnv, env->GetStringUTFChars(jEnv, taskClassName, JNI_FALSE));
    if (NULL == taskClass) {
        throwRuntimeException(jEnv, "获取任务类字节码失败");

        return -1;
    }
    /* 加锁 */
    pthread_mutex_lock(&g_ptask_pool->lock);

    if (g_ptask_pool->queue_size == g_ptask_pool->queue_max_size) {
        task_pool_print("任务队列已满，无法插入数据，请稍后重试");

        goto return1;
    }
    // queue_current的值只能由消费时的逻辑去操作，增加时不得操作，否则会出现数据混乱
    jint index = g_ptask_pool->queue_current % g_ptask_pool->queue_max_size;
    g_ptask_pool->task_cqueue[index] = taskClass;

    g_ptask_pool->queue_size++;
    /* 将加入的任务声明为全局引用，否则线程函数访问不到 */
    env->NewGlobalRef(jEnv, g_ptask_pool->task_cqueue[index]);

    task_pool_print("成功将任务加入任务池");

    /* 唤醒所有等待任务的线程 */
    pthread_cond_broadcast(&g_ptask_pool->cond_empty);

    /* 释放锁 */
    pthread_mutex_unlock(&g_ptask_pool->lock);

    return 0;

    return1:
    pthread_mutex_unlock(&g_ptask_pool->lock);

    return -1;
}

void task_pool_free() {
    assert(g_ptask_pool != NULL);

    print_nobuffer("【销毁任务池】开始释放任务池资源\n");

    pthread_mutex_destroy(&g_ptask_pool->lock);

    pthread_cond_destroy(&g_ptask_pool->cond_empty);

    free(g_ptask_pool->task_cqueue);
    free(g_ptask_pool);

    g_ptask_pool = NULL;
}

void task_pool_print(const char *msg) {
    assert(msg != NULL);

    char str[BUFSIZ] = {0};

    sprintf(str, "%d: %s: current=%d, size=%d, max_size=%d\n",
            (int) pthread_self(), msg, g_ptask_pool->queue_current, g_ptask_pool->queue_size,
            g_ptask_pool->queue_max_size);

    write(STDOUT_FILENO, str, strlen(str));
}